package com.quick.job;

import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author wangxin86
 * @title: JobLifecycle
 * @description: TODO
 * @date 2019/7/29 15:11
 */
@Component
public class JobLifecycle {
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    @Autowired
    private ApplicationContext applicationContext;

    @Resource
    private JobRegistryProperties jobRegistryProperties;

    private HashMap<String, SpringJobScheduler> jobSchedulerMap;

    @Autowired
    private DataSource dataSource;

    public void start(){
        this.startZookeeperRegistryCenter();
        this.regJob();
    }

    private void startZookeeperRegistryCenter(){
        zookeeperRegistryCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration(jobRegistryProperties.getServerList(),jobRegistryProperties.getNamespace()));
        zookeeperRegistryCenter.init();
    }


    private void regJob(){
        List<JobScheduleConfig> scheduleConfigs = this.getAllScheduleConfigs();
        if (null == scheduleConfigs || scheduleConfigs.size() == 0) {
            System.out.println("没有发现有可调度的任务");
            return;
        }

        jobSchedulerMap = new HashMap<>();

        for (JobScheduleConfig scheduleConfig : scheduleConfigs) {
            SpringJobScheduler jobScheduler = doRegJob(scheduleConfig);
            jobSchedulerMap.put(scheduleConfig.getJobName(), jobScheduler);
        }
    }



    private List<JobScheduleConfig> getAllScheduleConfigs(){
        Map<String, JobSchedulerRegistrar> registrarBeanMap = applicationContext.getBeansOfType(JobSchedulerRegistrar.class);
        if (null == registrarBeanMap) {
            return null;
        }

        return registrarBeanMap.values().stream()
                .flatMap(registrar -> Optional.ofNullable(registrar.getScheduleConfigs()).orElseGet(() -> Collections.emptyList()).stream())
                .collect(Collectors.toList());
    }

    private SpringJobScheduler doRegJob(JobScheduleConfig scheduleConfig){
        // 默认任务在单实例中运行，有其它需求再改
        JobCoreConfiguration coreConfiguration = JobCoreConfiguration
                .newBuilder(scheduleConfig.getJobName(), scheduleConfig.getScheduleCron(), jobRegistryProperties.getShardingTotalCount())
                .shardingItemParameters(jobRegistryProperties.getShardingItemParameters())
                .jobParameter(scheduleConfig.getJobParameter())
                .build();
        System.out.println("param="+scheduleConfig.getJobParameter());

        final SimpleJob jobExecutor = (SimpleJob) applicationContext.getBean(scheduleConfig.getJobExecutorName());

        SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(coreConfiguration, jobExecutor.getClass().getCanonicalName());
        // TODO 重写一个随机的Sharding策略，因为这里的任务都是单实例运行，为了避免所有的任务都压到同一个进程中去
        LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration
                .newBuilder(simpleJobConfiguration)
                .overwrite(true)
                .build();

        JobEventRdbConfiguration jobEventRdbConfig = new JobEventRdbConfiguration(dataSource);

        SpringJobScheduler jobScheduler = new SpringJobScheduler(
                (SimpleJob) shardingContext -> jobExecutor.execute(shardingContext),
                zookeeperRegistryCenter,
                liteJobConfiguration,
                jobEventRdbConfig);

        jobScheduler.init();
        System.out.println("任务【{"+scheduleConfig.getJobName()+"}】调度初始化完成, 调度表达式为【{"+scheduleConfig.getScheduleCron()+"}】");
        return jobScheduler;

    }
}
